package com.duowan.realtime.thrift;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.duowan.realtime.model.HyperloglogKey;
import com.duowan.realtime.uniq.algorithm.hyperloglog.HyperLogLog;

public class ProxyHyperLogLogServiceImpl {

	
	private static final String baseDir = "/data/backup/hyperloglog";
	private static final Logger logger = LoggerFactory.getLogger("hlllog");
	private HyperLogLogDB hllDB = null;
	
	public ProxyHyperLogLogServiceImpl(){
		hllDB = new HyperLogLogDB(baseDir);
	}

	private HyperLogLog chooseHllp(String hllGroup, String partition) {
		return hllDB.getHyperLogLogIfNotExistAndInit(hllGroup, partition);
	}

	public long offerForCardinality(String hllGroup,String group, List<String> values) {
		HyperLogLog hll = chooseHllp(hllGroup,group);
		return hll.offerForCardinality(values);
	}
	
	/**
	 * offer the value into hyperloglog pool
	 * @param hllGroup
	 * @param group
	 * @param values
	 */
	public void offer(String hllGroup,String group, List<String> values){
		HyperLogLog hll = chooseHllp(hllGroup,group);
		for (String value : values) {
			hll.offer(value);
		}
	}

	public long cardinality(String hllGroup, String group){
		HyperLogLog hll = chooseHllp(hllGroup,group);
		return hll.cardinality();
	}
	
	public List<Map<String, String>> keys(String hllGroup, String key){
		
		List<Map<String, String>> result = new ArrayList<Map<String,String>>();
		PartitionedHyperLogLog phll = hllDB.getPartitionedHyperLogLog(hllGroup);
		
		Set<String> keys = phll.getPartitons().keySet();
		for (String pkey : keys) {
			HyperloglogKey obj = null;
			try {
				if (pkey.startsWith(key)) {
					obj = HyperloglogKey.factory(pkey);
				}else {
					continue;
				}
			} catch (Throwable e) {
				logger.error("HyperloglogKey.factory error,the key is " + pkey, e);
				continue;
			}
			Map<String,Object> hyKeyMap = obj.toMap();
			logger.info("key to map:" + hyKeyMap);
			Iterator<Entry<String, Object>> it = hyKeyMap.entrySet().iterator();
			while (it.hasNext()) {
				if (it.next().getValue().toString().equals(key)) {
					it.remove();
				}
			}
			result.add((Map)hyKeyMap);
		}
		
		return result;
	}


	
}
